package com.coolingme.canal;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.concurrent.TimeUnit;

public class CanalClientExample {

    public static void main(String[] args) {
        // 连接到 Canal Server
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("192.168.0.162", 11111),
                "example", "root", "JsDyt@2020#$");

        connector.connect();
        connector.subscribe(".*\\..*"); // 订阅所有库的所有表

        System.out.println("开始监听 MySQL 的增删改操作...");

        while (true) {
            // 获取一批消息（默认最多1000条，超时2秒）
            Message message = connector.getWithoutAck(1000, 2000L, TimeUnit.MILLISECONDS);
            long batchId = message.getId();

            try {
                List<CanalEntry.Entry> entries = message.getEntries();
                if (batchId == -1 || entries.isEmpty()) {
                    System.out.println("无数据...");
                } else {
                    // 解析每一条 Entry
                    for (CanalEntry.Entry entry : entries) {
                        if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
                            parseRowData(entry);
                        }
                    }
                }

                // 确认消费成功
                connector.ack(batchId);

            } catch (Exception e) {
                System.err.println("处理消息失败: " + e.getMessage());
                connector.rollback(batchId); // 出错回滚
            }
        }
    }

    private static void parseRowData(CanalEntry.Entry entry) throws Exception {
        // 获取表名
        String tableName = entry.getHeader().getTableName();
        String schemaName = entry.getHeader().getSchemaName();

        // 解析行数据
        CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());

        for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
            CanalEntry.EventType eventType = rowChange.getEventType();

            System.out.println("事件类型: " + eventType.name() +
                    " | 库名: " + schemaName +
                    " | 表名: " + tableName);

            // 插入
            if (eventType == CanalEntry.EventType.INSERT) {
                System.out.println("新增数据:");
                rowData.getAfterColumnsList().forEach(col ->
                        System.out.println("  " + col.getName() + " = " + col.getValue())
                );
            }
            // 删除
            else if (eventType == CanalEntry.EventType.DELETE) {
                System.out.println("删除数据:");
                rowData.getBeforeColumnsList().forEach(col ->
                        System.out.println("  " + col.getName() + " = " + col.getValue())
                );
            }
            // 更新
            else if (eventType == CanalEntry.EventType.UPDATE) {
                System.out.println("更新数据:");
                rowData.getBeforeColumnsList().forEach(beforeCol -> {
                    CanalEntry.Column afterCol = rowData.getAfterColumnsList().stream()
                            .filter(c -> c.getName().equals(beforeCol.getName()))
                            .findFirst()
                            .orElse(null);

                    if (afterCol != null && !beforeCol.getValue().equals(afterCol.getValue())) {
                        System.out.println("  " + beforeCol.getName() +
                                " : " + beforeCol.getValue() + " → " + afterCol.getValue());
                    }
                });
            }
            System.out.println("----------------------------------------");
        }
    }
}
